package com.fxkj.cloud.drms.domain.ts.service;

import com.alicp.jetcache.Cache;
import com.alicp.jetcache.CacheManager;
import com.alicp.jetcache.anno.CacheType;
import com.alicp.jetcache.anno.Cached;
import com.alicp.jetcache.template.QuickConfig;
import com.fxkj.cloud.dbs.common.constants.CacheConstants;
import com.fxkj.cloud.dbs.common.model.DeviceAttributeKey;
import com.fxkj.cloud.dbs.common.model.kv.BasicTsKvEntry;
import com.fxkj.cloud.dbs.common.model.kv.TsKvEntry;
import com.fxkj.cloud.drms.domain.ts.repository.TimeseriesRepository;
import com.fxkj.common.converter.JetCacheDefaultKeyConverter;
import com.google.common.util.concurrent.FutureCallback;
import com.google.common.util.concurrent.Futures;
import com.google.common.util.concurrent.ListenableFuture;
import com.google.common.util.concurrent.MoreExecutors;
import org.springframework.stereotype.Service;

import java.time.Duration;
import java.util.*;
import java.util.stream.Collectors;

/**
 * <p>description: 时序数据查询服务</p>
 * @author ww
 * @date 2025/3/31 10:30
 */
@Service
public class TimeseriesServiceImpl implements TimeseriesService {
    private final TimeseriesRepository timeseriesRepository;
    /**
     * 设备属性缓存
     */
    private final Cache<UUID, Set<String>> deviceAttributeKeyCache;
    /**
     * 设备属性值缓存
     * key: deviceId+attributeKey
     */
    private final Cache<String, TsKvEntry> deviceAttributeValueCache;

    public TimeseriesServiceImpl(TimeseriesRepository timeseriesRepository, CacheManager cacheManager) {
        this.timeseriesRepository = timeseriesRepository;
        this.deviceAttributeKeyCache = cacheManager.getOrCreateCache(QuickConfig.newBuilder(CacheConstants.DeviceAttributeCache.DEVICE_ATTRIBUTE_DEVICE_ID_AND_ATTRIBUTE_ID_CACHE)
                .keyConvertor(new JetCacheDefaultKeyConverter())
                .localExpire(Duration.ZERO)
                .cacheType(CacheType.BOTH)
                .build());
        this.deviceAttributeValueCache = cacheManager.getOrCreateCache(QuickConfig.newBuilder(CacheConstants.DeviceAttributeCache.DEVICE_ATTRIBUTE_LAST_VALUE)
                .keyConvertor(new JetCacheDefaultKeyConverter())
                .localExpire(Duration.ofSeconds(30))
                .cacheType(CacheType.BOTH)
                .build());
    }

    @Override
    public ListenableFuture<List<TsKvEntry>> queryLastValueForString(UUID deviceId, List<String> attributeKeys) {
        //查询缓存
        Set<String> attributeKeySet = deviceAttributeKeyCache.get(deviceId);
        if (attributeKeySet != null && !attributeKeySet.isEmpty()) {
            //构建缓存key
            String deviceIdStr = deviceId.toString();
            List<TsKvEntry> result = new ArrayList<>();
            for (String attributeKey : attributeKeys) {
                String cacheKey = deviceIdStr + attributeKey;
                TsKvEntry cacheValue = deviceAttributeValueCache.get(cacheKey);
                if (cacheValue != null) {
                    result.add(cacheValue);
                }
            }
            if (!result.isEmpty()) {
                return Futures.immediateFuture(result);
            }
        }
        //查询数据库
        ListenableFuture<List<TsKvEntry>> future = timeseriesRepository.queryLastValueForString(deviceId, attributeKeys);
        Futures.addCallback(future, new FutureCallback<List<TsKvEntry>>() {
            @Override
            public void onSuccess(List<TsKvEntry> result) {
                if (result.isEmpty()) {
                    return;
                }
                //缓存设备键 todo可能存在分布式问题
                Set<String> attributeKeys = deviceAttributeKeyCache.computeIfAbsent(deviceId, a -> new HashSet<>());
                result.stream().map(TsKvEntry::getKey).forEach(attributeKeys::add);
                deviceAttributeKeyCache.put(deviceId, attributeKeys);

                //缓存设备最新键值
                //构建缓存key
                String deviceIdStr = deviceId.toString();
                Map<String, TsKvEntry> cacheValues = new HashMap<>();
                result.forEach(tskv -> cacheValues.put(deviceIdStr + tskv.getKey(), tskv));
                deviceAttributeValueCache.putAll(cacheValues);
            }

            @Override
            public void onFailure(Throwable t) {

            }
        }, MoreExecutors.directExecutor());
        return future;
    }
}
